#include "sys.h"
#include "log.h"
#include "thread_pool.h"
#include "cJSON.h"
#include "dial_strategy.h"

void dial_task_set_flag(Dial_Task_t* task, int flag)
{
    task->flag = flag;
}
void dial_task_all_set_flag(struct Dial_Task* taskall, int flag)
{
    struct Dial_Task_Head* head = &taskall->head;
    struct Dial_Task_s* ptr;
    TAILQ_FOREACH(ptr, head, entry)
    {
        dial_task_set_flag(ptr, flag);
    }
}

static void _parse_ping_tactics(cJSON* json, struct pingTactics_head* head)
{
    int i = 0, arr_len = 0;
    cJSON *json_obj = NULL, *json_ptr;
    pingTactics_t* ping_tactics;

    arr_len = cJSON_GetArraySize(json);
    for (i = 0; i < arr_len; i++) {
        ping_tactics = calloc(1, sizeof(pingTactics_t));
        json_obj = cJSON_GetArrayItem(json, i);
        json_ptr = cJSON_GetObjectItem(json_obj, "host");
        strcpy(ping_tactics->host, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "type");
        strcpy(ping_tactics->type, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "port");
        ping_tactics->port = json_ptr->valueint;
        json_ptr = cJSON_GetObjectItem(json_obj, "size");
        ping_tactics->size = json_ptr->valueint;
        json_ptr = cJSON_GetObjectItem(json_obj, "times");
        ping_tactics->times = json_ptr->valueint;
        json_ptr = cJSON_GetObjectItem(json_obj, "timeout");
        ping_tactics->timeout = json_ptr->valueint;
        json_ptr = cJSON_GetObjectItem(json_obj, "businessTag");
        strcpy(ping_tactics->businessTag, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "kafkaTopic");
        strcpy(ping_tactics->kafkaTopic, json_ptr->valuestring);
        TAILQ_INSERT_TAIL(head, ping_tactics, entry);
    }
}
static void _parse_tcp_tactics(cJSON* json, struct tcpTactics_head* head)
{
    int i = 0, arr_len = 0;
    cJSON *json_obj = NULL, *json_ptr;
    tcpTactics_t* tcp_tactics;

    arr_len = cJSON_GetArraySize(json);
    for (i = 0; i < arr_len; i++) {
        tcp_tactics = calloc(1, sizeof(tcpTactics_t));
        json_obj = cJSON_GetArrayItem(json, i);
        json_ptr = cJSON_GetObjectItem(json_obj, "tcpUrlVal");
        strcpy(tcp_tactics->host, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "port");
        json_ptr = cJSON_GetObjectItem(json_obj, "timeout");
        tcp_tactics->timeout = json_ptr->valueint;
        json_ptr = cJSON_GetObjectItem(json_obj, "businessTag");
        strcpy(tcp_tactics->businessTag, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "kafkaTopic");
        strcpy(tcp_tactics->kafkaTopic, json_ptr->valuestring);
        TAILQ_INSERT_TAIL(head, tcp_tactics, entry);
    }
}
static void _parse_dns_tactics(cJSON* json, struct dnsTactics_head* head)
{
    int i = 0, arr_len = 0;
    cJSON *json_obj = NULL, *json_ptr;
    dnsTactics_t* dns_tactics;

    arr_len = cJSON_GetArraySize(json);
    for (i = 0; i < arr_len; i++) {
        dns_tactics = calloc(1, sizeof(dnsTactics_t));
        json_obj = cJSON_GetArrayItem(json, i);
        json_ptr = cJSON_GetObjectItem(json_obj, "ip");
        strcpy(dns_tactics->ip, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "port");
        dns_tactics->port = json_ptr->valueint;
        json_ptr = cJSON_GetObjectItem(json_obj, "domainName");
        strcpy(dns_tactics->domainName, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "type");
        strcpy(dns_tactics->type, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "businessTag");
        strcpy(dns_tactics->businessTag, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "kafkaTopic");
        strcpy(dns_tactics->kafkaTopic, json_ptr->valuestring);
        TAILQ_INSERT_TAIL(head, dns_tactics, entry);
    }
}
static void _parse_http_tactics(cJSON* json, struct httpTactics_head* head)
{
    int i = 0, arr_len = 0;
    cJSON *json_obj = NULL, *json_ptr;
    httpTactics_t* http_tactics;

    arr_len = cJSON_GetArraySize(json);
    for (i = 0; i < arr_len; i++) {
        http_tactics = calloc(1, sizeof(httpTactics_t));
        json_obj = cJSON_GetArrayItem(json, i);
        json_ptr = cJSON_GetObjectItem(json_obj, "httpUrlVal");
        strcpy(http_tactics->url, json_ptr->valuestring);

        json_ptr = cJSON_GetObjectItem(json_obj, "port");
        http_tactics->port = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "timeout");
        http_tactics->timeout = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "businessTag");
        strcpy(http_tactics->businessTag, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "kafkaTopic");
        strcpy(http_tactics->kafkaTopic, json_ptr->valuestring);
        TAILQ_INSERT_TAIL(head, http_tactics, entry);
    }
}
static void _parse_traceroute_tactics(cJSON* json, struct tracerouteTactics_head* head)
{
    int i = 0, arr_len = 0;
    cJSON *json_obj = NULL, *json_ptr;
    tracerouteTactics_t* traceroute_tactics;

    arr_len = cJSON_GetArraySize(json);
    for (i = 0; i < arr_len; i++) {
        traceroute_tactics = calloc(1, sizeof(tracerouteTactics_t));
        json_obj = cJSON_GetArrayItem(json, i);
        json_ptr = cJSON_GetObjectItem(json_obj, "toAddress");
        strcpy(traceroute_tactics->toAddress, json_ptr->valuestring);

        json_ptr = cJSON_GetObjectItem(json_obj, "type");
        strcpy(traceroute_tactics->type, json_ptr->valuestring);

        json_ptr = cJSON_GetObjectItem(json_obj, "size");
        traceroute_tactics->size = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "maxNum");
        traceroute_tactics->maxNum = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "timeout");
        traceroute_tactics->timeout = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "businessTag");
        strcpy(traceroute_tactics->businessTag, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "kafkaTopic");
        strcpy(traceroute_tactics->kafkaTopic, json_ptr->valuestring);
        TAILQ_INSERT_TAIL(head, traceroute_tactics, entry);
    }
}
static void _parse_httpspeed_tactics(cJSON* json, struct httpspeedTactics_head* head, char* type)
{
    int i = 0, arr_len = 0;
    cJSON *json_obj = NULL, *json_ptr;
    httpspeedTactics_t* httpspeed_tactics;

    arr_len = cJSON_GetArraySize(json);
    for (i = 0; i < arr_len; i++) {
        httpspeed_tactics = calloc(1, sizeof(httpspeedTactics_t));
        json_obj = cJSON_GetArrayItem(json, i);

        json_ptr = cJSON_GetObjectItem(json_obj, "threads");
        httpspeed_tactics->threads = json_ptr->valueint;

        if (strcmp(type, "up") == 0)
            json_ptr = cJSON_GetObjectItem(json_obj, "upTime");
        else
            json_ptr = cJSON_GetObjectItem(json_obj, "downTime");
        httpspeed_tactics->totalTime = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "ignoreTime");
        httpspeed_tactics->ignoreTime = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "businessTag");
        strcpy(httpspeed_tactics->businessTag, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "kafkaTopic");
        strcpy(httpspeed_tactics->kafkaTopic, json_ptr->valuestring);
        TAILQ_INSERT_TAIL(head, httpspeed_tactics, entry);
    }
}
int dial_tactics_parse(Dial_Tactics_t* tactics, cJSON* json)
{
    cJSON* json_arr = NULL;

    TAILQ_INIT(&tactics->ping.head);
    json_arr = cJSON_GetObjectItem(json, "pingTactics");
    if (cJSON_IsArray(json_arr)) {
        pthread_mutex_init(&tactics->ping.mutex, NULL);
        pthread_cond_init(&tactics->ping.cond, NULL);
        tactics->ping.count = cJSON_GetArraySize(json_arr);
        _parse_ping_tactics(json_arr, &tactics->ping.head);
    }

    TAILQ_INIT(&tactics->tcp.head);
    json_arr = cJSON_GetObjectItem(json, "tcpTactics");
    if (cJSON_IsArray(json_arr)) {
        pthread_mutex_init(&tactics->tcp.mutex, NULL);
        pthread_cond_init(&tactics->tcp.cond, NULL);
        tactics->tcp.count = cJSON_GetArraySize(json_arr);
        _parse_tcp_tactics(json_arr, &tactics->tcp.head);
    }

    TAILQ_INIT(&tactics->dns.head);
    json_arr = cJSON_GetObjectItem(json, "dnsTactics");
    if (cJSON_IsArray(json_arr)) {
        pthread_mutex_init(&tactics->dns.mutex, NULL);
        pthread_cond_init(&tactics->dns.cond, NULL);
        tactics->dns.count = cJSON_GetArraySize(json_arr);
        _parse_dns_tactics(json_arr, &tactics->dns.head);
    }

    TAILQ_INIT(&tactics->http.head);
    json_arr = cJSON_GetObjectItem(json, "httpTactics");
    if (cJSON_IsArray(json_arr)) {
        pthread_mutex_init(&tactics->http.mutex, NULL);
        pthread_cond_init(&tactics->http.cond, NULL);
        tactics->http.count = cJSON_GetArraySize(json_arr);
        _parse_http_tactics(json_arr, &tactics->http.head);
    }

    TAILQ_INIT(&tactics->traceroute.head);
    json_arr = cJSON_GetObjectItem(json, "tracerouteTactics");
    if (cJSON_IsArray(json_arr)) {
        pthread_mutex_init(&tactics->traceroute.mutex, NULL);
        pthread_cond_init(&tactics->traceroute.cond, NULL);
        tactics->traceroute.count = cJSON_GetArraySize(json_arr);
        _parse_traceroute_tactics(json_arr, &tactics->traceroute.head);
    }

    TAILQ_INIT(&tactics->httpspeed.up_head);
    json_arr = cJSON_GetObjectItem(json, "httpUploadTactics");
    if (cJSON_IsArray(json_arr)) {
        pthread_mutex_init(&tactics->httpspeed.mutex, NULL);
        pthread_cond_init(&tactics->httpspeed.cond, NULL);
        tactics->httpspeed.count = cJSON_GetArraySize(json_arr);
        _parse_httpspeed_tactics(json_arr, &tactics->httpspeed.up_head, "up");
    }

    TAILQ_INIT(&tactics->httpspeed.down_head);
    json_arr = cJSON_GetObjectItem(json, "httpDownloadTactics");
    if (cJSON_IsArray(json_arr)) {
        pthread_mutex_init(&tactics->httpspeed.mutex, NULL);
        pthread_cond_init(&tactics->httpspeed.cond, NULL);
        tactics->httpspeed.count = cJSON_GetArraySize(json_arr);
        _parse_httpspeed_tactics(json_arr, &tactics->httpspeed.down_head, "down");
    }
    return 1;
}
static void _do_each_dial_tactics(struct Dial_Task_s* dialtask, Dial_Tactics_t* tactics, thread_pool_t* tpool)
{
    tactics->ping.taskId = dialtask->id;
    tactics->ping.version = dialtask->version;
    tactics->ping.kafka_url = dialtask->kafka_url;
    if (tactics->ping.count)
        thread_pool_add_task(tpool, do_ping_tactics, (void*)&tactics->ping);

    tactics->tcp.taskId = dialtask->id;
    tactics->tcp.version = dialtask->version;
    tactics->tcp.kafka_url = dialtask->kafka_url;
    if (tactics->tcp.count)
        thread_pool_add_task(tpool, do_tcp_tactics, (void*)&tactics->tcp);

    tactics->dns.taskId = dialtask->id;
    tactics->dns.version = dialtask->version;
    tactics->dns.kafka_url = dialtask->kafka_url;
    if (tactics->dns.count)
        thread_pool_add_task(tpool, do_dns_tactics, (void*)&tactics->dns);

    tactics->http.taskId = dialtask->id;
    tactics->http.version = dialtask->version;
    tactics->http.kafka_url = dialtask->kafka_url;
    if (tactics->http.count)
        thread_pool_add_task(tpool, do_http_tactics, (void*)&tactics->http);

    tactics->traceroute.taskId = dialtask->id;
    tactics->traceroute.version = dialtask->version;
    tactics->traceroute.kafka_url = dialtask->kafka_url;
    if (tactics->traceroute.count)
        thread_pool_add_task(tpool, do_traceroute_tactics, (void*)&tactics->traceroute);
}
static void _do_speed_tactics(struct Dial_Task_s* dialtask, httpspeed_tactics_t* httpspeed)
{
    httpspeed->taskId = dialtask->id;
    httpspeed->version = dialtask->version;
    httpspeed->kafka_url = dialtask->kafka_url;
    if (httpspeed->count)
        do_httpspeed_tactics(httpspeed, dialtask->nodeUrl, dialtask->probeId);
}
static void _do_each_tactics(struct Dial_Task_s* dialtask, Dial_Tactics_t* tactics)
{
    thread_pool_t* tpool;

    tpool = thread_pool_new(1);
    thread_pool_set_auto_extern(tpool);
    thread_pool_set_name(tpool, "each dial tactics");
    _do_each_dial_tactics(dialtask, tactics, tpool);
    thread_pool_wait(tpool);
    thread_pool_destroy(tpool);

    _do_speed_tactics(dialtask, &tactics->httpspeed);
}
static void* _start_task(void* arg)
{
    struct Dial_Task_s* dialtask;
    Dial_Tactics_t* tactics;
    time_t now, done_tm = 0;
    int flag = 0;

    dialtask = (struct Dial_Task_s*)arg;
    tactics = &dialtask->tactics;

    while (1) {

        sleep(1);
        now = time(NULL);

        switch (dialtask->flag) {
        case DIAL_TASK_FLAG_CANCEL:
            log_trace("Dial Task canceled.");
            goto out;
        case DIAL_TASK_FLAG_STOP:
            continue;
        case DIAL_TASK_FLAG_RUN:
        default:
            break;
        }

        if (now > dialtask->endTime) { //over time
            log_notice("Dial Task(id:%s, version:%s) is over time，now:%u, endTime:%u ",
                dialtask->id, dialtask->version, now, dialtask->endTime);
            goto out;
        }

        if (now < dialtask->startTime) {
            if (flag == 0) {
                log_warn("wait to startTime, now:%ld, startTime:%ld", now, dialtask->startTime);
                flag = 1;
            }
            continue;
        }

        if (now - done_tm < dialtask->collectTime) {
            continue;
        }

        if (now >= dialtask->startTime && now <= dialtask->endTime) {
            log_trace("Task(id:%s, version:%s)do each tactics", dialtask->id, dialtask->version);
            _do_each_tactics(dialtask, tactics);
            done_tm = time(NULL);
        }
    }

out:
    return NULL;
}
void* dial_task_run(void* arg)
{

    struct Dial_Task* taskall = (struct Dial_Task*)arg;
    struct Dial_Task_Head* head = &taskall->head;
    thread_pool_t* tpool = taskall->tpool;
    struct Dial_Task_s* dialtask;
    TAILQ_FOREACH(dialtask, head, entry)
    {
        thread_pool_add_task(tpool, _start_task, (void*)dialtask);
    }
    thread_pool_wait(tpool);
    return NULL;
}
// void dial_task_replace_by_id(struct Dial_Task_s* task, struct Dial_Task_Head* head)
// {
//     struct Dial_Task_s* ptr;

//     TAILQ_FOREACH(ptr, head, entry)
//     {
//         if (!strcmp(task->id, ptr->id)) {
//             TAILQ_REMOVE(head, ptr, entry);
//             dial_task_free(ptr);
//             TAILQ_INSERT_TAIL(head, task, entry);
//             break;
//         }
//     }
// }
void dial_task_free(struct Dial_Task_s* task)
{
    Dial_Tactics_t* tactics = &task->tactics;

    struct pingTactics* ping;
    TAILQ_FOREACH(ping, &tactics->ping.head, entry)
    {
        TAILQ_REMOVE(&tactics->ping.head, ping, entry);
        free(ping);
    }
    struct tcpTactics* tcp;
    TAILQ_FOREACH(tcp, &tactics->tcp.head, entry)
    {
        TAILQ_REMOVE(&tactics->tcp.head, tcp, entry);
        free(tcp);
    }
    struct dnsTactics* dns;
    TAILQ_FOREACH(dns, &tactics->dns.head, entry)
    {
        TAILQ_REMOVE(&tactics->dns.head, dns, entry);
        free(dns);
    }
    struct tracerouteTactics* traceroute;
    TAILQ_FOREACH(traceroute, &tactics->traceroute.head, entry)
    {
        TAILQ_REMOVE(&tactics->traceroute.head, traceroute, entry);
        free(traceroute);
    }
    struct httpTactics* http;
    TAILQ_FOREACH(http, &tactics->http.head, entry)
    {
        TAILQ_REMOVE(&tactics->http.head, http, entry);
        free(http);
    }
    struct httpspeedTactics* httpspeed;
    TAILQ_FOREACH(httpspeed, &tactics->httpspeed.down_head, entry)
    {
        TAILQ_REMOVE(&tactics->httpspeed.down_head, httpspeed, entry);
        free(httpspeed);
    }
    httpspeed = NULL;
    TAILQ_FOREACH(httpspeed, &tactics->httpspeed.up_head, entry)
    {
        TAILQ_REMOVE(&tactics->httpspeed.up_head, httpspeed, entry);
        free(httpspeed);
    }
    free(task);
}
void dial_task_free_all(struct Dial_Task* taskall)
{
    struct Dial_Task_Head* head = &taskall->head;
    thread_pool_t* tpool = taskall->tpool;
    struct Dial_Task_s* ptr;

    TAILQ_FOREACH(ptr, head, entry)
    {
        dial_task_set_flag(ptr, DIAL_TASK_FLAG_CANCEL);
    }
    thread_pool_wait(tpool);

    TAILQ_FOREACH(ptr, head, entry)
    {
        TAILQ_REMOVE(head, ptr, entry);
        dial_task_free(ptr);
    }
}